package com.zzw.demo.text_A;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zzw.demo.util.KafkaUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;


public class Text06 {
    public static void main(String[] args) throws Exception {

        //跳出率
        //6）在 DWM 层中，使用 CEP 筛选出跳出的数据明细，并在输出流中发送到相应的 Kafka 主题中（8分）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        //获取kafka消费者对象,读取数据封装流
        DataStreamSource<String> in = env.addSource(KafkaUtil.kafkaSource("ODS_BASE_LOG", "text06"));
        //TODO 4 对读取的数据进行类型转换 String -> JSONObject
        SingleOutputStreamOperator<JSONObject> map = in.map(x -> JSON.parseObject(x));
        // TODO 5.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> ts = map.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner((event, timestamp) -> event.getLong("ts")));
        // TODO 6 按照mid进行分组
        KeyedStream<JSONObject, String> jsonObjectStringKeyedStream = ts.keyBy(x -> x.getJSONObject("common").getString("mid"));
        // TODO 7 定义pattern
        Pattern<JSONObject, ?> pattern = Pattern.<JSONObject>begin("start").where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");
                        return StringUtils.isEmpty(last_page_id);
                    }
                }
        ).next("middle").subtype(JSONObject.class).where(
                new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String last_page_id = jsonObject.getJSONObject("page").getString("page_id");
                        return StringUtils.isEmpty(last_page_id);
                    }
                }
        ).within(Time.seconds(20));
        // TODO 8 将pattern应用到流上
        PatternStream<JSONObject> patternPS = CEP.pattern(jsonObjectStringKeyedStream, pattern);
        // TODO 9 从流中提取数据
        OutputTag<String> outputTag = new OutputTag<String>("CEP_Pattern") {};

        SingleOutputStreamOperator<String> first = patternPS.flatSelect(
                outputTag,
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map, long l, Collector<String> collector) throws Exception {
                        List<JSONObject> first = map.get("first");
                        for (JSONObject jsonObj : first) {
                            collector.collect(jsonObj.toJSONString());
                        }
                    }
                },
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {

                    }
                }
            );

        DataStream<String> sideOutput = first.getSideOutput(outputTag);
        // TODO 10 将跳出明细写到kafka的dwm层主题
        sideOutput.addSink(KafkaUtil.kafkaProducer("dwm_user_jump_detail"));

        //执行程序
        env.execute();

    }
}
